package com.zhouyu.mysql;

import cn.hutool.core.collection.CollUtil;
import cn.hutool.core.io.FileUtil;
import cn.hutool.core.io.file.LineSeparator;
import cn.hutool.json.JSONUtil;
import com.zhouyu.mysql.core.*;
import com.zhouyu.mysql.dict.*;
import com.zhouyu.mysql.page.IndexPage;
import net.sf.jsqlparser.schema.Table;
import net.sf.jsqlparser.statement.alter.Alter;
import net.sf.jsqlparser.statement.alter.AlterExpression;
import net.sf.jsqlparser.statement.create.table.CreateTable;
import net.sf.jsqlparser.statement.create.table.Index;
import net.sf.jsqlparser.statement.insert.Insert;
import net.sf.jsqlparser.statement.select.FromItemVisitorAdapter;
import net.sf.jsqlparser.statement.select.PlainSelect;
import net.sf.jsqlparser.statement.select.SelectItem;

import java.io.BufferedReader;
import java.io.File;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;

import static com.zhouyu.mysql.page.IndexPage.PAGE_NEW_INFIMUM;

/**
 * 大都督周瑜（我的微信: dadudu6789）
 */
public class SqlExecutor {

    private HaInnodb haInnodb = new HaInnodb();

    public void createTable(CreateTable createTableStatement) {
        haInnodb.createTable(createTableStatement);
    }

    public void insert(Insert insertStatement) {
        haInnodb.insert(insertStatement);
    }

    public String select(PlainSelect plainSelectStatement, boolean isAll, boolean isSecondary, int rootPageNo) {

        // 获取表名
        final String[] tableNameArray = new String[1];
        plainSelectStatement.getFromItem().accept(new FromItemVisitorAdapter() {
            @Override
            public void visit(Table table) {
                tableNameArray[0] = table.getName();
            }
        });
        String tableName = tableNameArray[0];

        DictTable dictTable = SystemDict.getInstance().getNameTables().get(tableName);

        DTuple dTuple = DTupleUtil.convert(plainSelectStatement);

        List<String> selectColumnList = new ArrayList<>();
        for (SelectItem<?> selectItem : plainSelectStatement.getSelectItems()) {
            selectColumnList.add(selectItem.toString());
        }

        List<HashMap<String, Object>> result = select(selectColumnList, dTuple, dictTable, isAll, isSecondary, rootPageNo);

        return result.toString();
    }

    /**
     *
     * @param dTuple        走索引时的where条件，如果是全表扫描则不关心
     * @param dictTable     当前表
     * @param isAll         是不是全表扫描
     * @param isSecondary   是不是走的辅助索引
     * @param rootPageNo    走索引时，对应索引的根页页号
     * @return 查询结果
     */
    private List<HashMap<String, Object>> select(List<String> selectColumnList, DTuple dTuple, DictTable dictTable, boolean isAll, boolean isSecondary, int rootPageNo) {
        PageCur pageCur = new PageCur();
        if (isAll) {
            // 如果是要全表扫描，就取当前表对应的B+树的最左边叶子节点
            pageCur.setPageNo(BtrUtil.getFirstLeafPage(dictTable).fil_page_get_page_offset());
            pageCur.setRec(PAGE_NEW_INFIMUM);
        } else {
            // 走索引，则从对应索引的根页开始
            pageCur.setPageNo(rootPageNo);
            pageCur.setRec(PAGE_NEW_INFIMUM);
        }

        List<HashMap<String, Object>> result = new ArrayList<>();

        while (true) {
            SelectOneResult selectOneResult = haInnodb.select_one(dictTable, pageCur, dTuple, isAll, isSecondary);
            boolean endOfIndex = selectOneResult.isEndOfIndex();

            if (endOfIndex) {
                break;
            }

            // 如果是全表扫描或走的主键索引，则能直接拿出所有字段，如果走的是辅助索引，则需要进行回表才能拿出所有字段
            HashMap<String, Object> recContent = selectOneResult.getRecContent();
            if (recContent != null) {
                if (isSecondary) {

                    // 如果select的字段在recContent中都有，则可以直接返回，否则需要回表
                    if (CollUtil.isNotEmpty(selectColumnList) && recContent.keySet().containsAll(selectColumnList)) {
                        HashMap<String, Object> resultRecContent = new HashMap<>();
                        for (String selectColumnName : selectColumnList) {
                            resultRecContent.put(selectColumnName, recContent.get(selectColumnName));
                        }
                        result.add(resultRecContent);
                    } else {
                        DTuple idTuple = new DTuple();
                        DField dField = new DField();
                        dField.setDictColumn(DictColumn.builder().name("id").type("int").len(4).build());
                        dField.setData(recContent.get("id"));
                        idTuple.getFields().add(dField);

                        // 回表，就是用id取主键索引上找出结果，所以不是全表扫描，不是辅助索引
                        result.addAll(select(null, idTuple, dictTable, false, false, dictTable.getPageNo()));
                    }

                } else {
                    result.add(recContent);
                }

            }

            pageCur = selectOneResult.getPageCur();
        }

        return result;
    }

    public void alter(Alter alter) {
        Table table = alter.getTable();
        DictTable dictTable = SystemDict.getInstance().getNameTables().get(table.getName());

        List<AlterExpression> alterExpressions = alter.getAlterExpressions();
        Index index = alterExpressions.get(0).getIndex();

        DictIndex dictIndex = new DictIndex();
        dictIndex.setIndexName(index.getName());
        dictIndex.setIndexId(SystemDict.getInstance().addMaxIndexId());
        dictIndex.setTableId(dictTable.getTableId());

        // 一个索引定义了有哪些字段
        List<String> fieldNames = new ArrayList<>();
        for (Index.ColumnParams column : index.getColumns()) {
            String columnName = column.getColumnName();
            DictColumn dictColumn = dictTable.getDictColumnList().stream().filter(c -> c.getName().equals(columnName)).findFirst().get();
            DictField dictField = new DictField();
            dictField.setDictColumn(dictColumn);
            dictField.setPrefixLen(dictColumn.getLen());  // 写死，暂时不支持前缀索引
            dictIndex.getDictFieldList().add(dictField);
            fieldNames.add(columnName);
        }

        // 记录一个表有哪些索引
        List<DictIndex> dictIndices = SystemDict.getInstance().getTableIdIndexes().get(dictTable.getTableId());
        if (CollUtil.isEmpty(dictIndices)) {
            dictIndices = new ArrayList<>();
        }
        dictIndices.add(dictIndex);
        SystemDict.getInstance().getTableIdIndexes().put(dictTable.getTableId(), dictIndices);

        // 先在表空间中开辟新页作为索引的根页
        int rootIndexPageNo = PageUtil.createPage(dictTable.getSpaceId());
        dictIndex.setPageNo(rootIndexPageNo);
        IndexPage rootIndexPage = PageUtil.readPage(dictTable.getSpaceId(), rootIndexPageNo);

        // 持久化到系统字典中
        SystemDict.getInstance().serialize();


        // 再扫描聚集索引上的数据，将索引字段对应的数据取出来放入buf中
        List<BufEntry> buf = new ArrayList<>(); // buf本质上就是一块内存，List也是内存
        int bufMax = 4;
        List<File> tempFiles = new ArrayList<>();
        File tmpDir = FileUtil.file("tmp"); // 用来存临时文件的目录

        // 全表扫描
        DTuple emptyTuple = new DTuple();
        PageCur pageCur = new PageCur();
        pageCur.setPageNo(BtrUtil.getFirstLeafPage(dictTable).fil_page_get_page_offset());
        pageCur.setRec(PAGE_NEW_INFIMUM);
        while (true) {
            SelectOneResult selectOneResult = haInnodb.select_one(dictTable, pageCur, emptyTuple, true, false);
            boolean endOfIndex = selectOneResult.isEndOfIndex();

            if (endOfIndex) {
                break;
            }

            // 某条记录内容
            HashMap<String, Object> recContent = selectOneResult.getRecContent();

            if (recContent != null) {

                // 构造一个Entry，存入buf用来排序，后续再存到索引中，用fields进行排序
                BufEntry bufEntry = new BufEntry();
                bufEntry.setId((Integer) recContent.get("id"));
                for (String fieldName : fieldNames) {
                    bufEntry.getFields().add((String) recContent.get(fieldName));
                }
                buf.add(bufEntry);

                if (buf.size() >= bufMax) {
                    // 如果buf满，则先排序，并先存到临时文件中，并清空buf
                    // 需要利用bufEntry中对应的索引字段进行排序，需要比较大小，这里跟对应字段的大小规则有关系，我们只跟

                    CollUtil.sort(buf, (entry1, entry2) -> {

                        int size = entry1.getFields().size();

                        for (int i = 0; i < size; i++) {
                            // 如果两个字段不相等，则比较直接返回这两个字段的大小关系，否则继续比较第二个字段
                            if (!entry1.getFields().get(i).equals(entry2.getFields().get(i))) {
                                return entry1.getFields().get(i).compareTo(entry2.getFields().get(i));
                            }
                        }

                        // 所有字段都相等则返回0
                        return 0;
                    });

                    // 存到临时文件
                    // 得按自定义格式来存，不能直接把buf序列化到文件中，因为如果后面都记录List来进行归并排序，那么越往后List会越大，那不符合需求了
                    // 需求是一个List是有固定大小的，是由于一个List放不下所有数据才用文件排序，所以文件排序时不能再把所有数据又加回到内存中
                    // 只能按自定义格式一条一条数据写到文件，后面一条一条数据从文件读取出来进行排序
                    File tempFile = FileUtil.createTempFile(tmpDir);
                    for (BufEntry entry : buf) {
                        FileUtil.appendUtf8String(JSONUtil.toJsonStr(entry), tempFile);
                        FileUtil.appendUtf8String(LineSeparator.MAC.getValue(), tempFile);
                    }
                    tempFiles.add(tempFile);

                    // 清空buf
                    buf.clear();
                }
            }

            pageCur = selectOneResult.getPageCur();
        }

        // 数据都遍历完了，先检查前面是否生成了临时文件，如果生成了临时文件，则需要进行归并排序，并且把最终的排序结果要插入到索引中
        if (CollUtil.isNotEmpty(tempFiles)) {

            // 如果生成了临时文件，但是数据遍历完了之后，最后一个buf没有满，此时也需要把最后一个buf生成临时文件，一起进行归并排序
            if (CollUtil.isNotEmpty(buf)) {
                File tempFile = FileUtil.createTempFile(tmpDir);
                for (BufEntry entry : buf) {
                    FileUtil.appendUtf8String(JSONUtil.toJsonStr(entry), tempFile);
                    FileUtil.appendUtf8String(LineSeparator.MAC.getValue(), tempFile);
                }
                tempFiles.add(tempFile);

                // 清空buf
                buf.clear();
            }

            try {
                File sortResult = FileSort.sort(tmpDir);

                BufferedReader reader = FileUtil.getUtf8Reader(sortResult);
                String s1 = reader.readLine();
                while (s1 != null) {
                    BufEntry bufEntry = JSONUtil.toBean(s1, BufEntry.class);
                    // 将bufEntry插入到索引
                    DTuple dTuple = DTupleUtil.convert(bufEntry);
                    PageUtil.insert_row(rootIndexPage, dTuple);
                    s1 = reader.readLine();
                }

                sortResult.delete();
            } catch (Exception e) {
                throw new RuntimeException(e);
            }
        } else {
            CollUtil.sort(buf, (entry1, entry2) -> {

                int size = entry1.getFields().size();

                for (int i = 0; i < size; i++) {
                    // 如果两个字段不相等，则比较直接返回这两个字段的大小关系，否则继续比较第二个字段
                    if (!entry1.getFields().get(i).equals(entry2.getFields().get(i))) {
                        return entry1.getFields().get(i).compareTo(entry2.getFields().get(i));
                    }
                }

                // 所有字段都相等则返回0
                return 0;
            });

            for (BufEntry bufEntry : buf) {
                // 将bufEntry插入到索引
                DTuple dTuple = DTupleUtil.convert(bufEntry);
                PageUtil.insert_row(rootIndexPage, dTuple);
            }
        }
    }
}
